package org.example.rocketmqstudy.producer;

import io.netty.channel.DefaultChannelId;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class OrderlyProducer {
    public static AtomicLong idGenerator = new AtomicLong(1000);

    public static void main(String[] args) throws InterruptedException, MQClientException, MQBrokerException, RemotingException {
        DefaultChannelId.newInstance(); // 解决 RemotingTimeoutException: invokeSync call timeout 异常

        DefaultMQProducer producer = new DefaultMQProducer("orderly-producer");
        producer.setNamesrvAddr("8.138.36.56:9876");
        producer.start();

        for (int i = 0; i < 5; i++) {
            createOrder(producer);
        }

        producer.shutdown();
    }

    public static void createOrder(DefaultMQProducer producer) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        // 模拟全局ID生成
        long orderId = idGenerator.incrementAndGet();

        // 发送订单生成消息
        Message orderCreateMsg = new Message("test-order", ("订单生成，orderId=" + orderId).getBytes());
        sendOrderlyMessage(producer, orderCreateMsg, orderId);

        // 发送支付消息
        Message payMsg = new Message("test-order", ("支付，orderId=" + orderId).getBytes());
        sendOrderlyMessage(producer, payMsg, orderId);

        // 发送发货消息
        Message shipMsg = new Message("test-order", ("发货，orderId=" + orderId).getBytes());
        sendOrderlyMessage(producer, shipMsg, orderId);
    }

    public static void sendOrderlyMessage(DefaultMQProducer producer, Message message, long orderId) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long orderId = (Long) arg;
                int index = (int)(orderId % mqs.size());
                return mqs.get(index);
            }
        }, orderId);

        String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        System.out.printf("%s [%s] 发送消息，msg：%s %n", now, Thread.currentThread().getName(), new String(message.getBody()));
    }
}
